package com.danbay.framework.protocol.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 *
 * @author SJG
 * 2017/9/27
 */
@Slf4j
public class MqttClient {

    private final MqttProperties config;

    private org.eclipse.paho.client.mqttv3.MqttClient client;

    private com.danbay.framework.protocol.mqtt.MqttCallback mqttCallback;

    public MqttClient(MqttProperties mqttProperties) {
        this.config = mqttProperties;
    }

    /**
     *  连接服务器
     */
    private void connect() throws MqttException {
        if(client!=null){
            client.disconnect();
            //重新初始化连接
            init();
        }
        // MemoryPersistence设置clientid的保存形式，默认为以内存保存
        client = new org.eclipse.paho.client.mqttv3.MqttClient(
                config.getHost()+":"+config.getPort(),
                config.getClientId()+"_"+ System.currentTimeMillis(),
                new MemoryPersistence()
        );
        MqttConnectOptions options = new MqttConnectOptions();
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(config.isCleanSession());
        options.setUserName(config.getUserName());
        options.setPassword(config.getPassWord().toCharArray());
        // // 设置超时时间 单位为秒
        options.setConnectionTimeout(config.getConnectionTimeout());
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机
        options.setKeepAliveInterval(config.getKeepAliveInterval());
        options.setAutomaticReconnect(true);
        // 设置回调
        client.setCallback(mqttCallback);
        client.connect(options);
    }


    /**
     * 发布消息
     * @param topic 主题名
     * @param msg 消息
     * @throws MqttException 发布异常
     */
    public void publish(String topic , String msg) throws MqttException {
        MqttTopic mqttTopic= client.getTopic(topic);
        MqttMessage message = new MqttMessage();
        message.setQos(config.getDefaultQos());
        message.setPayload(msg.getBytes());

        mqttTopic.publish(message);
    }

    /**
     * 订阅消息
     * @param topic 主题名
     * @throws MqttException 订阅异常
     */
    public void subscribe(String topic) throws MqttException {
        int[] qos  = {config.getDefaultQos()};
        String[] topic1 = {topic};
        client.subscribe(topic1, qos);
    }

    /**
     * 一次订阅多个消息
     * @param topics 主题名
     * @throws MqttException 订阅异常
     */
    private void subscribe(String[] topics) throws MqttException {
        int[] qos  = new int[topics.length];
        for(int i=0;i<qos.length;i++){
            qos[i]=config.getDefaultQos();
        }

        client.subscribe(topics, qos);
    }

    /**
     * 初始化
     * @throws MqttException mqtt异常
     */
    public void init(MqttCallback mqttCallback) throws MqttException {
        try {
            this.mqttCallback = mqttCallback;
            this.connect();
            //启动后自动订阅主题
            String[] topics = config.getSubscribes();
            if (topics != null && topics.length > 0 && !"".equals(topics[0])) {
                subscribe(topics);
            }
        }catch (MqttException e){
            //延时5秒重试连接
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e1) {
                log.error("thread sleep error",e1);
            }
            init();
            throw e;
        }
    }

    private void init() throws MqttException {
        init(this.mqttCallback);
    }
}
